#include "netlib/rpc/rpc_channel_impl.h"

#include "netlib/base/logger.h"
#include "rpc.pb.h"

namespace netlib::rpc {

void RpcChannelImpl::CallMethod(const ::google::protobuf::MethodDescriptor* method,
                                ::google::protobuf::RpcController* controller,
                                const ProtobufMessage* request,
                                ProtobufMessage* response,
                                ::google::protobuf::Closure* done) {
	LOG_DEBUG << "Client CallMethod: " << method->name();
	RpcMessage msg;
	msg.set_type(kRequest);
	auto id = id_creator_.fetch_add(1);
	msg.set_id(id);
	msg.set_service(method->service()->full_name());
	msg.set_method(method->name());
	msg.set_request(request->SerializeAsString());

	{
		MutexLockGuard gd(mtx_);
		req_res_map_.emplace(id, std::make_pair(response, done));
	}
	codec_->Send(conn_, msg);
}
void RpcChannelImpl::OnProtobufMessage(const net::TcpConnectionPtr& conn,
                                       std::unique_ptr<ProtobufMessage>&& msg,
                                       Timestamp ts) {
	LOG_DEBUG << "Client receive RpcMessage: " << msg->DebugString();
	auto rmsg = dynamic_cast<RpcMessage*>(msg.get());
	assert(rmsg->type() == kResponse);
	assert(rmsg->has_response());
	RequestResult res{nullptr, nullptr};
	{
		MutexLockGuard gd(mtx_);
		auto it = req_res_map_.find(rmsg->id());
		if (it != req_res_map_.end()) {
			res = it->second;
			req_res_map_.erase(it);
		}
	}
	assert(res.first);
	auto success = res.first->ParseFromString(rmsg->response());
	assert(success);
	if (res.second) {
		res.second->Run();
	}
}
} // namespace netlib::rpc